package com.atguigu.api4

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

/**
 * @description: 流式输出
 * @time: 2020/7/22 17:22
 * @author: baojinlong
 **/
object TableExample {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 从文本读取
    // val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")
    // 从流中读取
    val inputStreamFromFile: DataStream[String] = environment.socketTextStream("localhost", 7777)


    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })

    // 创建表执行环境
    val tableEnvironment: StreamTableEnvironment = StreamTableEnvironment.create(environment)
    // 基于数据流,转换成一张表然后进行操作
    val dataTable: Table = tableEnvironment.fromDataStream(dataStream)
    // 调用TableApi得到转换结果
    val resultTable: Table = dataTable.select("id,temperature")
      .filter("id == 'sensor_01'")

    // 加上下面一句话就可以直接 select id,temperature from dataTable where id='sensor_01'
    // tableEnvironment.createTemporaryView("dataTable",dataTable)
    // 依然是流式输出,直接写sql得到转换结果
    val resultSql: Table = tableEnvironment.sqlQuery("select id,temperature from " + dataTable + " where id='sensor_01'")
    // 转换成流打印输出
    val resultStream: DataStream[(String, Double)] = resultSql.toAppendStream[(String, Double)]
    // val resultStream: DataStream[(String, Double)] = resultTable.toAppendStream[(String, Double)]
    resultStream.print("result-table")

    resultTable.printSchema()
    environment.execute("table example job")
  }

}
